
package org.shana.shortlink.core.config;

import org.shana.shortlink.core.mq.consumer.ShortLinkStatsSaveConsumer;
import lombok.RequiredArgsConstructor;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;

import java.time.Duration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static org.shana.shortlink.core.common.constant.RedisKeyConstant.SHORT_LINK_STATS_STREAM_GROUP_KEY;
import static org.shana.shortlink.core.common.constant.RedisKeyConstant.SHORT_LINK_STATS_STREAM_TOPIC_KEY;
@Configuration
@RequiredArgsConstructor
public class RedisStreamConfiguration {

    // Redis连接工厂，用于创建Redis连接
    private final RedisConnectionFactory redisConnectionFactory;

    // Redis Stream消息消费者，用于处理Redis Stream中的消息
    private final ShortLinkStatsSaveConsumer shortLinkStatsSaveConsumer;

    // 定义一个线程池，用于异步消费Redis Stream中的消息
    @Bean
    public ExecutorService asyncStreamConsumer() {
        // 使用原子整型计数器来给线程命名
        AtomicInteger index = new AtomicInteger();
        // 获取可用处理器数量
        int processors = Runtime.getRuntime().availableProcessors();
        // 创建线程池
        return new ThreadPoolExecutor(
                processors,                                     // 核心线程数为可用处理器数量
                processors + processors >> 1,                   // 最大线程数为核心线程数的1.5倍
                60,                                             // 线程空闲时间为60秒
                TimeUnit.SECONDS,                               // 空闲时间单位为秒
                new LinkedBlockingQueue<>(),                    // 使用无界队列来存储等待执行的任务
                runnable -> {                                   // 线程工厂，用于创建线程
                    Thread thread = new Thread(runnable);       // 创建线程
                    // 给线程命名，格式为：stream_consumer_short-link_stats_{index}
                    thread.setName("stream_consumer_short-link_stats_" + index.incrementAndGet());
                    // 设置线程为守护线程
                    thread.setDaemon(true);
                    return thread;                              // 返回创建的线程
                }
        );
    }

    // 创建StreamMessageListenerContainer实例，用于监听Redis Stream中的消息并进行消费
    @Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer(ExecutorService asyncStreamConsumer) {
        // 配置消息监听器容器选项
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                        .builder()
                        .batchSize(10)                              // 一次最多获取10条消息
                        .executor(asyncStreamConsumer)             // 使用异步消费者线程池来执行消费任务
                        .pollTimeout(Duration.ofSeconds(3))        // 设置拉取消息的超时时间为3秒
                        .build();

        // 创建StreamMessageListenerContainer实例
        StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer =
                StreamMessageListenerContainer.create(redisConnectionFactory, options);

        // 设置消息的自动确认机制，并指定消费者组、消息消费的起始位置和消息处理器
        streamMessageListenerContainer.receiveAutoAck(
                Consumer.from(SHORT_LINK_STATS_STREAM_GROUP_KEY, "stats-consumer"),  // 消费者组和消费者ID
                StreamOffset.create(SHORT_LINK_STATS_STREAM_TOPIC_KEY, ReadOffset.lastConsumed()),  // 消息消费的起始位置
                shortLinkStatsSaveConsumer  // 消息处理器
        );

        return streamMessageListenerContainer;  // 返回StreamMessageListenerContainer实例
    }
}
